/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.ban.ycl.wx.nrd.update.util;

import cn.ac.iie.di.ban.ycl.wx.nrd.update.exception.UpdateException;
import com.mongodb.BasicDBObject;
import com.mongodb.MongoClient;
import com.mongodb.MongoCredential;
import com.mongodb.QueryOperators;
import com.mongodb.ServerAddress;
import com.mongodb.bulk.BulkWriteResult;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoCursor;
import com.mongodb.client.MongoDatabase;
import com.mongodb.client.model.UpdateOneModel;
import com.mongodb.client.model.UpdateOptions;
import com.mongodb.client.model.WriteModel;
import java.lang.reflect.Array;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.log4j.Logger;
import org.bson.Document;

/**
 *
 * @author zwlin
 */
public class MongoHelper {

	private static final Logger logger = Logger.getLogger(MongoHelper.class.getName());

	MongoClient conn;
	MongoDatabase db;
	public MongoCollection<Document> collection;

	String[] retFieldNames;

	public MongoHelper(String[] hosts, String dbname, String tablename, String username, String passwd) throws UpdateException {
		try {
			List<ServerAddress> seeds = new ArrayList<>();
			for (String host : hosts) {
				String[] serv = host.split(":");
				if (serv != null && serv.length == 2) {
					seeds.add(new ServerAddress(serv[0], Integer.parseInt(serv[1])));
				}
			}
			MongoCredential credential = MongoCredential.createScramSha1Credential(username, dbname, passwd.toCharArray());
			List<MongoCredential> credentialsList = new ArrayList<MongoCredential>();
			credentialsList.add(credential);

			conn = new MongoClient(seeds, credentialsList);
			db = conn.getDatabase(dbname);
			collection = db.getCollection(tablename);
		} catch (Exception ex) {
			try {
				conn.close();
			} catch (Exception ex1) {
			}
			throw new UpdateException("cacnnot init mongo helper !", ex);

		}
	}

	public static Document getDocument(HashMap<String, Object> rec) {
		Document doc = new Document();
		for (Map.Entry<String, Object> entry : rec.entrySet()) {
			Object v = entry.getValue();
			if (v.getClass().isArray()) {
				int n = Array.getLength(v);
				List l = new ArrayList(n);
				for (int i = 0; i < n; ++i) {
					l.add(Array.get(v, i));
				}
				doc.put(entry.getKey(), l);
			} else {
				doc.put(entry.getKey(), v);
			}
		}
		return doc;
	}

	/**
	 * 批量查询
	 *
	 * @param keyCol	用来定位记录的列名
	 * @param keys	用来定位记录的列值集合
	 * @param retFields	返回的列名
	 * @return
	 */
	public List<HashMap<String, Object>> find(String keyCol, Collection<String> keys, String[] retFields) {
		List<HashMap<String, Object>> rtn = new ArrayList<>(keys.size());
		MongoCursor<Document> cursor = collection.find(new Document(keyCol, new BasicDBObject(QueryOperators.IN, keys))).iterator();
		while (cursor.hasNext()) {
			Document rec = cursor.next();
			HashMap<String, Object> rtnRec = new HashMap<>(retFields.length);
			for (String str : retFields) {
				rtnRec.put(str, rec.get(str));
			}
			rtn.add(rtnRec);
		}
		return rtn;
	}

	/**
	 * 批量更新，将某一列的值增加一
	 *
	 * @param keyCol	用于定位记录的列名
	 * @param keys	用于定位记录的列值集合
	 * @param incCol	需要加一的列名
	 */
	public void upinc(String keyCol, Collection<String> keys, String incCol) {
		List<WriteModel<Document>> updatedoc = new ArrayList<>();
		for (String key : keys) {
			Document doc = new Document();
			doc.put(incCol, 1L);
			updatedoc.add(
					new UpdateOneModel<Document>(new Document(keyCol, key),
							new Document("$inc", doc), new UpdateOptions().upsert(false)));
		}
		collection.bulkWrite(updatedoc);
	}

	/**
	 * 批量更新，将某一列的值增加一定数值。
	 *
	 * @param keyCol	用于定位记录的列名
	 * @param updateCol	用于增加的列名
	 * @param updateValues	keyCol的值和updateCol的增量的对应关系
	 */
	public void upadd(String keyCol, String updateCol, HashMap<String, Long> updateValues) {
		List<WriteModel<Document>> updatedoc = new ArrayList<>();
		for (Map.Entry<String, Long> entry : updateValues.entrySet()) {
			Document doc = new Document();
			doc.put(updateCol, entry.getValue());
			updatedoc.add(
					new UpdateOneModel<Document>(new Document(keyCol, entry.getKey()),
							new Document("$inc", doc), new UpdateOptions().upsert(false)));
		}
		BulkWriteResult bulkWrite = collection.bulkWrite(updatedoc);
		int a = 0;
	}

	/**
	 * 批量更新，将某一列的值更新为新值。
	 *
	 * @param keyCol	用于定位记录的列名
	 * @param updateCol	用于更新的列名
	 * @param updateValues	keyCol的值和updateCol的增量的对应关系
	 */
	public void update(String keyCol, String updateCol, HashMap<String, Object> updateValues) {
		List<WriteModel<Document>> updatedoc = new ArrayList<>();
		for (Map.Entry<String, Object> entry : updateValues.entrySet()) {
			Document doc = new Document();
			doc.put(updateCol, entry.getValue());
                        doc.put("ut", (long)System.currentTimeMillis()/1000);
                        logger.info(entry.getKey()+"---"+(long)System.currentTimeMillis()/1000);
			updatedoc.add(
					new UpdateOneModel<Document>(new Document(keyCol, entry.getKey()),
							new Document("$set", doc), new UpdateOptions().upsert(false)));
		}
		BulkWriteResult bulkWriteResult= collection.bulkWrite(updatedoc);
                logger.info("update to mongo " + bulkWriteResult.getModifiedCount());
	}
        
        
        
        public void update(String keyCol, HashMap<String/*update_keycol_value*/, HashMap<String, Object>> updateMap) {
		List<WriteModel<Document>> updatedoc = new ArrayList<>();
		for (Map.Entry<String, HashMap<String, Object>> entry : updateMap.entrySet()) {
//                    logger.info(entry.getKey()+"---"+entry.getValue());
			Document doc = new Document();
			doc.putAll(entry.getValue());
			updatedoc.add(
					new UpdateOneModel<Document>(new Document(keyCol, entry.getKey()),
							new Document("$set", doc), new UpdateOptions().upsert(false)));
		}
		BulkWriteResult bulkWrite=collection.bulkWrite(updatedoc);
//                logger.info("update to mpngodb "+bulkWrite.getMatchedCount());
                logger.info("update to mpngo "+bulkWrite.getModifiedCount());
	}
}
